package com.youku.collector;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class Consumer implements Task<Record>, Counter {

	private BlockingQueue<Record> queue;

	public Consumer(BlockingQueue<Record> queue) {
		super();
		this.queue = queue;
	}

	private long start;

	private long end;

	private AtomicInteger count = new AtomicInteger(0);

	@Override
	public void run() {
		while (true) {
			try {
				Record record = take();
				handle(record);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	private void handle(Record record) {
		switch (record.getRole()) {
		case "start":
			start = System.currentTimeMillis();
			break;
		case "end":
			end = System.currentTimeMillis();
			System.out.println("cost " + (end - start) + " ms and sender count is " + count);
			break;
		default:
			send(record);
			break;
		}
	}

	private void send(Record record) {
		try {
			incr();
			System.out.println(record);
			Thread.sleep(50);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	@Override
	public int count() {
		return count.get();
	}

	@Override
	public void incr() {
		count.incrementAndGet();
	}

	@Override
	public BlockingQueue<Record> getQueue() {
		return queue;
	}

	@Override
	public void put(Record element) {
		throw new UnsupportedMethodException("消费者不支持put方法");
	}

	@Override
	public Record take() throws InterruptedException {
		return queue.take();
	}

}
